// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: fengyang

#include <memory>
#include <fmt/format.h>

#include <boost/bimap.hpp>
#include <boost/assign.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/http.hpp>
#include <boost/beast/ssl/ssl_stream.hpp>

#include "coin2/exchange/base/feed/rest/market_data_client.h"

#include "coin2/base/time.h"
#include "coin2/base/hostname_util.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "coin2/app/driver.h"
#include "coin2/exchange/di/constant.h"
#include "coin2/exchange/base/feed/ws/subscriber_manager.h"
#include "coin2/strategy/strategy.h"
#include "coin2/exchange/factory/feed/raw_processor.h"
#include "coin2/exchange/base/executor/topic_distributor.h"

namespace coin2::exchange::base::feed::rest {

using coin2::exchange::base::executor::TopicRecordPacket;
using coin2::exchange::base::executor::TopicRecordProcessor;
using coin2::exchange::base::feed::FeedUpdate;
using coin2::exchange::base::feed::FeedParsingProcessor;
using coin2::exchange::base::feed::SubscriberManager;
using coin2::exchange::base::symbology::IProduct;
using coin2::exchange::di::ConstantInjection;
using coin2::exchange::di::ConstantInjectionBuilder;
using coin2::app::DriverConfig;
using coin2::exchange::feed::WebsocketConfig;
using fastfeed::proto::KlineInterval;
using fastfeed::proto::Feed;

namespace {

const char* config_template = R"(
{{
    "feed": {{
        "common": {{
            "type": "FEED_SYSTEM_TYPE_LIVE",
            "websocket": {{
                "{feed_type}_start": "{start}",
                "{feed_type}_end": "{end}"
            }},
        }},
        "exchanges": {{
            "{mea}": {{
                "channels": ["{channel}"],
                "products": {{
                    "norms": ["{norm}"]
                }}
            }},
        }}
    }}
}}
)";

const char* binance_spot_override_config_template = R"(
{{
    "feed": {{
        "common": {{
            "type": "FEED_SYSTEM_TYPE_LIVE",
            "websocket": {{
                "kline_start": "{start}",
                "kline_end": "{end}"
            }},
        }},
        "exchanges": {{
            "{mea}": {{
                "channels": ["{channel}"],
                "products": {{
                    "norms": ["{norm}"]
                }},
                "websocket": {{
                    "urls": {{
                        "kline_rest": {{
                            "rest_host": "https://feed-proxy-jp-01.joomo.io/binance-api",
                            "unauthorized_limit": {{
                                "subscription_rate": "1200/PT1M"
                            }}
                        }}
                    }}
                }}
            }}
        }}
    }}
}}
)";

const char* binance_futures_override_config_template = R"(
{{
    "feed": {{
        "common": {{
            "type": "FEED_SYSTEM_TYPE_LIVE",
            "websocket": {{
                "kline_start": "{start}",
                "kline_end": "{end}"
            }},
        }},
        "exchanges": {{
            "{mea}": {{
                "channels": ["{channel}"],
                "products": {{
                    "norms": ["{norm}"]
                }},
                "websocket": {{
                    "urls": {{
                        "kline_rest": {{
                            "rest_host": "https://feed-proxy-jp-01.joomo.io/binance-fapi",
                            "unauthorized_limit": {{
                                "subscription_rate": "1200/PT1M"
                            }}
                        }}
                    }}
                }}
            }}
        }}
    }}
}}
)";

DriverConfig BuildConfig(
    const std::string& mea,
    const std::string& norm,
    const std::string& channel,
    const std::string& start,
    const std::string& end,
    const ConstantInjection& ci,
    const DriverConfig* rate_config,
    const std::string& feed_type) {
  std::string tmpl = config_template;
  auto hostname = GetLocalHostname();
  if (feed_type == "kline") {
    if (hostname.find("us-east-1") != std::string::npos) {
      if (mea == "Spot.Binance.v1") {
        tmpl = binance_spot_override_config_template;
      } else if (mea == "Futures.Binance.v1") {
        tmpl = binance_futures_override_config_template;
      }
    }
  }
  std::string config_json = fmt::format(tmpl,
                                        fmt::arg("mea", mea),
                                        fmt::arg("norm", norm),
                                        fmt::arg("channel", channel),
                                        fmt::arg("start", start),
                                        fmt::arg("end", end),
                                        fmt::arg("feed_type", feed_type));

  DriverConfig config;
  auto status = google::protobuf::util::JsonStringToMessage(config_json, &config);
  CHECK_THROW(status.ok()) << "Fail to parse DriverConfig from \n"
                           << config_json
                           << status;

  if (rate_config && rate_config->feed().exchanges().count(mea) > 0) {
    const auto& config_exchange = rate_config->feed().exchanges().at(mea);
    auto& exchange = config.mutable_feed()->mutable_exchanges()->at(mea);
    if (exchange.has_websocket() && exchange.websocket().urls_size() > 0) {
      // No need to override.
    } else {
      exchange.MergeFrom(config_exchange);
      LOG_FIRST_N(INFO, 1) << mea << " subscription rate config ovrride";
    }
  }
  ConsolidateDriverConfig(ci, &config);
  return config;
};

std::unique_ptr<IProduct> BuildProduct(
    const std::string& mea,
    const DriverConfig& config,
    int64_t resovle_time) {
  auto products =  ProductRequestsFromConfig(
      mea,
      config.feed().exchanges().begin()->second.products(),
      resovle_time,
      true);  // die_on_failure
  CHECK_EQ(products.size(), 1);
  return std::move(*products.begin());
}

std::unique_ptr<SubscriberManager> BuildSubscriber(
    const MarketExchangeApi& mea,
    const ConstantInjection& ci,
    const DriverConfig& config) {
  std::unique_ptr<SubscriberManager> subscriber = std::make_unique<SubscriberManager>(
      ci.GetFeedSubscriptionSpec(mea),
      config.feed().exchanges().begin()->second.websocket(),
      coin2::flow::TopicType::SHORT_TOPIC,
      nullptr,  // executor
      config.feed().exchanges().begin()->second.raw_parser().enable_telemetry());
  return subscriber;
}

std::unique_ptr<FeedParsingProcessor> BuildProcessor(
    const ConstantInjection& ci,
    const std::string& system_name,
    const MarketExchangeApi& mea,
    const DriverConfig& config) {
  return coin2::exchange::factory::feed::NewRawProcessor(
      ci,
      system_name,
      mea,
      config.feed().exchanges().begin()->second.raw_parser(),
      config.feed().exchanges().begin()->second.recipe());
}

void RegisterProduct(
    const MarketExchangeApi& mea,
    std::unique_ptr<IProduct>& prod,
    std::unique_ptr<SubscriberManager>& subscriber,
    std::unique_ptr<FeedParsingProcessor>& processor,
    coin2::strategy::IStrategy* strategy) {
    auto symbol = prod->feed_native_symbol();
    char group = coin2::flow::SymbolGroupConfig::GetInstance().GetGroupForRelativeNorm(
        mea,
        prod->relative_norm());
    subscriber->RequestNativeSymbol(symbol, group);
    processor->AddProduct(*prod, strategy);  // to strategy
}

std::vector<std::string> BuildTarget(
    const MarketExchangeApi& mea,
    std::unique_ptr<SubscriberManager>& subscriber,
    const WebsocketConfig& config,
    const std::vector<WsConnectionConfig>& connection_configs) {
  auto generator =  factory::feed::NewRequestGenerator(mea, &config);
  std::vector<std::string> targets;
  for (const WsConnectionConfig& config : connection_configs) {
    std::vector<std::string> reqs = generator->GenerateRequests(
        config.url.channels,
        config.symbols,
        config.mea_config->channels,
        config.url.request);
    std::vector<std::string> urls;
    for (const auto& req : reqs) {
        urls.push_back(config.url.path + req);
    }
    targets.insert(targets.end(), urls.begin(), urls.end());
  }

  return targets;
}

class Connection {
 public:
  Connection(
    const std::string& host,
    const std::string& port)
    : port_(port),
      ssl_ctx_(boost::asio::ssl::context::sslv23) {
    // only support https
    CHECK_EQ(host.rfind(url_prefix_), 0);
    host_ = host.substr(url_prefix_.size());
    stream_ssl_.reset(new boost::beast::ssl_stream<boost::beast::tcp_stream>(
        ioc_, ssl_ctx_));
    CHECK_THROW(SSL_set_tlsext_host_name(stream_ssl_->native_handle(), host_.c_str()))
        << "SNI error " << host_ << " " << port_;
    boost::asio::ip::tcp::resolver resolver(ioc_);
    auto const results = resolver.resolve(host_, port_);
    boost::beast::get_lowest_layer(*stream_ssl_).expires_after(
        std::chrono::nanoseconds(timeout_ns_));
    boost::beast::get_lowest_layer(*stream_ssl_).socket().connect(*results.begin());
    stream_ssl_->handshake(boost::asio::ssl::stream_base::client);
  }

  bool Get(const std::string& target, std::string* content) {
    boost::beast::http::request<boost::beast::http::string_body> request;
    const int version = 11;
    request.version(version);
    request.method(boost::beast::http::verb::get);
    request.keep_alive(true);
    request.set(boost::beast::http::field::host, host_);
    // coinbase rest require user agent
    request.set(boost::beast::http::field::user_agent, "Beast");
    request.target(boost::beast::string_view(target.data(), target.size()));
    boost::beast::http::write(*stream_ssl_, request);

    boost::beast::flat_buffer buffer;
    boost::beast::http::response<boost::beast::http::string_body> response;
    boost::beast::http::read(*stream_ssl_, buffer, response);
    boost::beast::http::status status = response.result();
    *content = response.body();

    return boost::beast::http::status::ok == status;
  }


 private:
  std::string host_;
  std::string port_;
  boost::asio::io_context ioc_;
  boost::asio::ssl::context ssl_ctx_;
  std::unique_ptr<boost::beast::ssl_stream<boost::beast::tcp_stream>> stream_ssl_;

  // Timeout config
  static constexpr int64_t timeout_ns_ = 60 * 1000000000L;
  static constexpr std::string_view url_prefix_ = "https://";
};

std::shared_ptr<TopicRecordPacket> BuildPacket(
    const std::string& topic,
    const std::string& payload,
    bool wrap_response,
    const std::string& host,
    const std::string& target) {
  std::string data = payload;
  if (wrap_response) {
    data = fmt::format("{{\"url\":\"{}{}\", \"data\":{}}}", host, target, payload);
  }

  auto packet = std::shared_ptr<TopicRecordPacket>(
      new TopicRecordPacket{
          topic,
          0,
          TopicRecordProcessor::GetTimestamp(),
          0,
          0,
          0,
          0,
          0,
          0,
          data});
  return packet;
}

void SendRequest(
    const MarketExchangeApi& mea,
    std::unique_ptr<SubscriberManager>& subscriber,
    const WebsocketConfig& config,
    const std::vector<WsConnectionConfig>& connection_configs,
    std::unique_ptr<FeedParsingProcessor>& processor) {
  int64_t num = connection_configs.front().url.unauthorized_limit.subscription->first;
  int64_t window = connection_configs.front().url.unauthorized_limit.subscription->second;
  int64_t interval = window / num + 1;
  std::string host = connection_configs.front().url.host;
  std::string port = connection_configs.front().url.port;
  std::string topic = connection_configs.front().topic;
  bool wrap_response = connection_configs.front().url.wrap_response;
  std::vector<std::string> targets = BuildTarget(
      mea, subscriber, config, connection_configs);

  std::unique_ptr<Connection> connection = std::make_unique<Connection>(
      host, port);
  bool first_send = true;
  for (auto& target : targets) {
    if (!first_send) {
      std::this_thread::sleep_for(std::chrono::nanoseconds(interval));
    } else {
      first_send = false;
    }
    std::string response;
    if (!connection->Get(target, &response)) {
      LOG(ERROR)<< "Qeury kline failed: "<< host << ":" << port << target
        << ", response: " << response;
      continue;
    }
    if (response.size() < 1000) {
      std::cout << response << std::endl;
    }
    auto packet = BuildPacket(topic, response, wrap_response, host, target);
    processor->OnTopicRecord(*packet.get());
  }
}

class KlineFeedReader : public coin2::strategy::IStrategy {
 public:
  explicit KlineFeedReader(
    int64_t start,
    int64_t end)
    : start_(start),
      end_(end) {
  }

  void onKlineFeed(const FeedUpdate& upd) override {
    auto& klines = upd.klines();
    auto& proto = proto_;

    auto symbol = upd.product().relative_norm();
    if (symbol_) {
      CHECK_EQ(symbol_.value(), symbol) << "Only support 1 symbol";
    } else {
      symbol_ = symbol;
    }

    auto interval = (*klines.Get())->interval;
    if (interval_) {
      CHECK_EQ(interval_.value(), interval) << "Only support 1 interval";
    } else {
      interval_ = interval;
      proto.Clear();
    }


    proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_KLINE);
    proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_KLINE);
    proto.set_symbol(symbol);
    proto.set_timestamp(upd.timestamp());
    proto.set_origin_timestamp(klines.Get().value()->timestamp);

    for (auto iter = klines.RBegin(); iter != klines.REnd(); iter++) {
      if (iter->open_timestamp < start_ ||
          iter->open_timestamp >= end_ ||
          iter->open_timestamp <= last_open_timestamp_) {
        continue;
      }

      auto* pkline = proto.mutable_kline_snapshot()->add_kline();
      pkline->set_interval(iter->interval);
      pkline->set_open_timestamp(iter->open_timestamp);
      pkline->set_close_timestamp(iter->close_timestamp);
      pkline->set_open(iter->open);
      pkline->set_close(iter->close);
      pkline->set_high(iter->high);
      pkline->set_low(iter->low);
      if (!std::isnan(iter->volume)) pkline->set_volume(iter->volume);
      if (!std::isnan(iter->turnover)) pkline->set_turnover(iter->turnover);
      if (!std::isnan(iter->buy_volume)) pkline->set_buy_volume(iter->buy_volume);
      if (!std::isnan(iter->buy_turnover)) pkline->set_buy_turnover(iter->buy_turnover);

     last_open_timestamp_ = iter->open_timestamp;
    }
  }

  const Feed& GetFeed() {
    return proto_;
  }

 private:
  int64_t start_;
  int64_t end_;
  int64_t last_open_timestamp_ = 0;
  std::optional<std::string> symbol_;
  std::optional<KlineInterval> interval_;
  mutable Feed proto_;
};

class TradingDataFeedReader : public coin2::strategy::IStrategy {
 public:
  explicit TradingDataFeedReader(
    int64_t start,
    int64_t end)
    : start_(start),
      end_(end) {
  }

  void onTopLongShortAccountRatioFeed(const FeedUpdate& upd) override {
    auto& trading_data = upd.top_long_short_account_ratio();
    auto& proto = proto_;

    auto symbol = upd.product().relative_norm();
    if (symbol_) {
      CHECK_EQ(symbol_.value(), symbol) << "Only support 1 symbol";
    } else {
      symbol_ = symbol;
    }

    auto interval = trading_data.Interval();;
    if (interval_) {
      CHECK_EQ(interval_.value(), interval) << "Only support 1 interval";
    } else {
      interval_ = interval;
      proto.Clear();
    }


    proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_TRADING_DATA);
    proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_TRADING_DATA);
    proto.set_symbol(symbol);
    proto.set_timestamp(upd.timestamp());
    proto.set_origin_timestamp(trading_data.Timestamp());

    for (auto iter = trading_data.RBegin(); iter != trading_data.REnd(); iter++) {
      if (iter->timestamp < start_ ||
          iter->timestamp >= end_ ||
          iter->timestamp <= last_timestamp_) {
        continue;
      }

      auto* pratio = proto.mutable_trading_data_snapshot()->add_top_long_short_account_ratio();
      pratio->set_timestamp(iter->timestamp);
      pratio->set_long_short_ratio(iter->long_short_ratio);
      pratio->set_long_(iter->long_account);
      pratio->set_short_(iter->short_account);

      last_timestamp_ = iter->timestamp;
    }
  }

  void onTopLongShortPositionRatioFeed(const FeedUpdate& upd) override {
    auto& trading_data = upd.top_long_short_position_ratio();
    auto& proto = proto_;

    auto symbol = upd.product().relative_norm();
    if (symbol_) {
      CHECK_EQ(symbol_.value(), symbol) << "Only support 1 symbol";
    } else {
      symbol_ = symbol;
    }

    auto interval = trading_data.Interval();;
    if (interval_) {
      CHECK_EQ(interval_.value(), interval) << "Only support 1 interval";
    } else {
      interval_ = interval;
      proto.Clear();
    }


    proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_TRADING_DATA);
    proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_TRADING_DATA);
    proto.set_symbol(symbol);
    proto.set_timestamp(upd.timestamp());
    proto.set_origin_timestamp(trading_data.Timestamp());

    for (auto iter = trading_data.RBegin(); iter != trading_data.REnd(); iter++) {
      if (iter->timestamp < start_ ||
          iter->timestamp >= end_ ||
          iter->timestamp <= last_timestamp_) {
        continue;
      }

      auto* pratio = proto.mutable_trading_data_snapshot()->add_top_long_short_position_ratio();
      pratio->set_timestamp(iter->timestamp);
      pratio->set_long_short_ratio(iter->long_short_ratio);
      pratio->set_long_(iter->long_position);
      pratio->set_short_(iter->short_position);

      last_timestamp_ = iter->timestamp;
    }
  }

  void onGlobalLongShortAccountRatioFeed(const FeedUpdate& upd) override {
    auto& trading_data = upd.global_long_short_account_ratio();
    auto& proto = proto_;

    auto symbol = upd.product().relative_norm();
    if (symbol_) {
      CHECK_EQ(symbol_.value(), symbol) << "Only support 1 symbol";
    } else {
      symbol_ = symbol;
    }

    auto interval = trading_data.Interval();;
    if (interval_) {
      CHECK_EQ(interval_.value(), interval) << "Only support 1 interval";
    } else {
      interval_ = interval;
      proto.Clear();
    }


    proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_TRADING_DATA);
    proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_TRADING_DATA);
    proto.set_symbol(symbol);
    proto.set_timestamp(upd.timestamp());
    proto.set_origin_timestamp(trading_data.Timestamp());

    for (auto iter = trading_data.RBegin(); iter != trading_data.REnd(); iter++) {
      if (iter->timestamp < start_ ||
          iter->timestamp >= end_ ||
          iter->timestamp <= last_timestamp_) {
        continue;
      }

      auto* pratio = proto.mutable_trading_data_snapshot()->add_global_long_short_account_ratio();
      pratio->set_timestamp(iter->timestamp);
      pratio->set_long_short_ratio(iter->long_short_ratio);
      pratio->set_long_(iter->long_account);
      pratio->set_short_(iter->short_account);

      last_timestamp_ = iter->timestamp;
    }
  }

  void onTakerBuySellRatioFeed(const FeedUpdate& upd) override {
    auto& trading_data = upd.taker_buy_sell_ratio();
    auto& proto = proto_;

    auto symbol = upd.product().relative_norm();
    if (symbol_) {
      CHECK_EQ(symbol_.value(), symbol) << "Only support 1 symbol";
    } else {
      symbol_ = symbol;
    }

    auto interval = trading_data.Interval();;
    if (interval_) {
      CHECK_EQ(interval_.value(), interval) << "Only support 1 interval";
    } else {
      interval_ = interval;
      proto.Clear();
    }


    proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_TRADING_DATA);
    proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_TRADING_DATA);
    proto.set_symbol(symbol);
    proto.set_timestamp(upd.timestamp());
    proto.set_origin_timestamp(trading_data.Timestamp());

    for (auto iter = trading_data.RBegin(); iter != trading_data.REnd(); iter++) {
      if (iter->timestamp < start_ ||
          iter->timestamp >= end_ ||
          iter->timestamp <= last_timestamp_) {
        continue;
      }

      auto* pratio = proto.mutable_trading_data_snapshot()->add_taker_buy_sell_ratio();
      pratio->set_timestamp(iter->timestamp);
      pratio->set_buy_sell_ratio(iter->buy_sell_ratio);
      pratio->set_buy_volume(iter->buy_volume);
      pratio->set_sell_volume(iter->sell_volume);

      last_timestamp_ = iter->timestamp;
    }
  }

  void onOpenInterestHistFeed(const FeedUpdate& upd) override {
    auto& trading_data = upd.open_interest_hist();
    auto& proto = proto_;

    auto symbol = upd.product().relative_norm();
    if (symbol_) {
      CHECK_EQ(symbol_.value(), symbol) << "Only support 1 symbol";
    } else {
      symbol_ = symbol;
    }

    auto interval = trading_data.Interval();;
    if (interval_) {
      CHECK_EQ(interval_.value(), interval) << "Only support 1 interval";
    } else {
      interval_ = interval;
      proto.Clear();
    }


    proto.set_feed_type(fastfeed::proto::FeedType::FEED_TYPE_TRADING_DATA);
    proto.set_data_type(fastfeed::proto::DataType::DATA_TYPE_TRADING_DATA);
    proto.set_symbol(symbol);
    proto.set_timestamp(upd.timestamp());
    proto.set_origin_timestamp(trading_data.Timestamp());

    for (auto iter = trading_data.RBegin(); iter != trading_data.REnd(); iter++) {
      if (iter->timestamp < start_ ||
          iter->timestamp >= end_ ||
          iter->timestamp <= last_timestamp_) {
        continue;
      }

      auto* open_interest_hist = proto.mutable_trading_data_snapshot()->add_open_interest_hist();
      open_interest_hist->set_timestamp(iter->timestamp);
      open_interest_hist->set_qty_float(iter->open_interest_qty);

      last_timestamp_ = iter->timestamp;
    }
  }

  const Feed& GetFeed() {
    return proto_;
  }

 private:
  int64_t start_;
  int64_t end_;
  int64_t last_timestamp_ = 0;
  std::optional<std::string> symbol_;
  std::optional<TradingDataInterval> interval_;
  mutable Feed proto_;
};

bool IntervalChannelConverter(KlineInterval* interval, std::string* channel) {
  typedef boost::bimap<KlineInterval, std::string> BiMapKlineChanel;
  static BiMapKlineChanel bimap =  boost::assign::list_of<BiMapKlineChanel::relation>
      (KlineInterval::KLINE_INTERVAL_1MIN, "kline_rest_1min")
      (KlineInterval::KLINE_INTERVAL_3MIN, "kline_rest_3min")
      (KlineInterval::KLINE_INTERVAL_5MIN, "kline_rest_5min")
      (KlineInterval::KLINE_INTERVAL_15MIN, "kline_rest_15min")
      (KlineInterval::KLINE_INTERVAL_30MIN, "kline_rest_30min")
      (KlineInterval::KLINE_INTERVAL_1HOUR, "kline_rest_1hour")
      (KlineInterval::KLINE_INTERVAL_2HOUR, "kline_rest_2hour")
      (KlineInterval::KLINE_INTERVAL_4HOUR, "kline_rest_4hour")
      (KlineInterval::KLINE_INTERVAL_8HOUR, "kline_rest_8hour")
      (KlineInterval::KLINE_INTERVAL_12HOUR, "kline_rest_12hour")
      (KlineInterval::KLINE_INTERVAL_1DAY, "kline_rest_1day")
      (KlineInterval::KLINE_INTERVAL_1WEEK, "kline_rest_1week")
      (KlineInterval::KLINE_INTERVAL_1MON, "kline_rest_1month");

  if(*interval == KlineInterval::KLINE_INTERVAL_INVALID) {
    // convert channel to interval enum
    auto iter = bimap.right.find(*channel);
    if (iter != bimap.right.end()) {
      *interval = iter->second;
      return true;
    } else {
      return false;
    }
  } else if (channel->empty()) {
    // convert inteval to channel
    auto iter = bimap.left.find(*interval);
    if (iter != bimap.left.end()) {
      *channel = iter->second;
      return true;
    } else {
      return false;
    }
  }

  return false;
}

bool TradingDataIntervalChannelConverter(TradingDataInterval* interval, std::string* channel) {
  typedef boost::bimap<TradingDataInterval, std::string> BiMapTradingDataChanel;
  static BiMapTradingDataChanel bimap =  boost::assign::list_of<BiMapTradingDataChanel::relation>
      (TradingDataInterval::TRADING_DATA_INTERVAL_1MIN, "1min")
      (TradingDataInterval::TRADING_DATA_INTERVAL_3MIN, "3min")
      (TradingDataInterval::TRADING_DATA_INTERVAL_5MIN, "5min")
      (TradingDataInterval::TRADING_DATA_INTERVAL_15MIN, "15min")
      (TradingDataInterval::TRADING_DATA_INTERVAL_30MIN, "30min")
      (TradingDataInterval::TRADING_DATA_INTERVAL_1HOUR, "1hour")
      (TradingDataInterval::TRADING_DATA_INTERVAL_2HOUR, "2hour")
      (TradingDataInterval::TRADING_DATA_INTERVAL_4HOUR, "4hour")
      (TradingDataInterval::TRADING_DATA_INTERVAL_8HOUR, "8hour")
      (TradingDataInterval::TRADING_DATA_INTERVAL_12HOUR, "12hour")
      (TradingDataInterval::TRADING_DATA_INTERVAL_1DAY, "1day")
      (TradingDataInterval::TRADING_DATA_INTERVAL_1WEEK, "1week")
      (TradingDataInterval::TRADING_DATA_INTERVAL_1MON, "1month");

  if(*interval == TradingDataInterval::TRADING_DATA_INTERVAL_INVALID) {
    // convert channel to interval enum
    auto iter = bimap.right.find(*channel);
    if (iter != bimap.right.end()) {
      *interval = iter->second;
      return true;
    } else {
      return false;
    }
  } else if (channel->empty()) {
    // convert inteval to channel
    auto iter = bimap.left.find(*interval);
    if (iter != bimap.left.end()) {
      *channel = iter->second;
      return true;
    } else {
      return false;
    }
  }

  return false;
}

void CheckInterval(
    const ConstantInjection& ci,
    const MarketExchangeApi& mea,
    const std::string& interval_channel) {
  const auto& spec = ci.GetFeedSubscriptionSpec(mea);
  const auto& channels = spec.channels;
  std::string kline_rest("kline_rest_");
  bool found = false;
  std::vector<std::string> valid_intervals;
  for (const auto& [key, _] : channels) {
    std::string channel_name = key;
    if (channel_name == interval_channel) {
      found = true;
      break;
    } else if (StringStartsWith(channel_name, kline_rest)) {
      KlineInterval interval_val = KlineInterval::KLINE_INTERVAL_INVALID;
      CHECK(IntervalChannelConverter(&interval_val, &channel_name));
      std::string interval_name = KlineInterval_Name(interval_val);
      valid_intervals.push_back(interval_name);
    }
  }
  CHECK(found) << ", " << interval_channel.substr(kline_rest.size()) <<
      " interval is not supported, do you mean " << Join(valid_intervals);
}

void TradingDataCheckInterval(
    const ConstantInjection& ci,
    const MarketExchangeApi& mea,
    const std::string& interval_channel) {
  const auto& spec = ci.GetFeedSubscriptionSpec(mea);
  const auto& channels = spec.channels;
  std::string trading_data_rest("trading_data_rest_");
  bool found = false;
  std::vector<std::string> valid_channels;
  for (const auto& [key, _] : channels) {
    std::string channel_name = key;
    if (channel_name == interval_channel) {
      found = true;
      break;
    } else if (StringStartsWith(channel_name, trading_data_rest)) {
      valid_channels.push_back(channel_name);
    }
  }
  CHECK(found) << ", " << interval_channel.substr(trading_data_rest.size()) <<
      " channel is not supported, do you mean " << Join(valid_channels);
}

}  // namespace

MarketDataClient::MarketDataClient()
    : ci_(ConstantInjectionBuilder(), true) {
  std::experimental::filesystem::path subscription_rate_file =
      "data/coin2/feed/kline_subscription_rate.json";
  auto content = ReadFileContent(subscription_rate_file);
  if (content.empty()) {
    LOG(INFO) << "Empty subscription rate config, use default";
  } else {
    bool success = JsonStringToProto(content, &rate_config_);
    if (!success) {
      LOG(ERROR) << "Fail to parse product info to proto: " << subscription_rate_file;
    } else {
      rate_config_enable_ = true;
    }
  }
}

Feed MarketDataClient::QueryKlines(
    const std::string& mea_str,
    const std::string& norm,
    KlineInterval interval,
    int64_t start,
    int64_t end) const {
  start = start / 1'000'000'000 * 1'000'000'000;
  end = end / 1'000'000'000 * 1'000'000'000;
  CHECK_GT(end, start) << "Invalid time duration " << start << "-" << end;
  KlineFeedReader kline_reader(start, end);
  auto mea = MarketExchangeApi::FromString(mea_str);
  std::string interval_channel;
  CHECK(IntervalChannelConverter(&interval, &interval_channel));
  CheckInterval(ci_, mea, interval_channel);
  std::string start_str = Iso8601FromTimestamp(start);
  std::string end_str = Iso8601FromTimestamp(end);
  const DriverConfig* rate_config = nullptr;
  if (rate_config_enable_) {
    rate_config = &rate_config_;
  }
  auto config = BuildConfig(mea_str, norm, interval_channel, start_str, end_str, ci_, rate_config, "kline");
  auto prod = BuildProduct(mea_str, config, start);
  CHECK(prod) << "Invliad symbol " << norm;
  auto subscriber = BuildSubscriber(mea, ci_, config);
  auto processor = BuildProcessor(
      ci_, config.feed().exchanges().begin()->first, mea, config);
  RegisterProduct(mea, prod, subscriber, processor, &kline_reader);
  auto connection_configs = subscriber->GenerateConnectionConfigs();
  SendRequest(mea, subscriber, config.feed().exchanges().begin()->second.websocket(),
      connection_configs, processor);

  return kline_reader.GetFeed();
}

Feed MarketDataClient::QueryTopLongShortAccountRatio(
    const std::string& mea_str,
    const std::string& norm,
    TradingDataInterval interval,
    int64_t start,
    int64_t end) const {
  return QueryTradingData(mea_str, norm, interval, start, end, "trading_data_rest_top_account");
}

Feed MarketDataClient::QueryTopLongShortPositionRatio(
    const std::string& mea_str,
    const std::string& norm,
    TradingDataInterval interval,
    int64_t start,
    int64_t end) const {
  return QueryTradingData(mea_str, norm, interval, start, end, "trading_data_rest_top_position");
}

Feed MarketDataClient::QueryGlobalLongShortAccountRatio(
    const std::string& mea_str,
    const std::string& norm,
    TradingDataInterval interval,
    int64_t start,
    int64_t end) const {
  return QueryTradingData(mea_str, norm, interval, start, end, "trading_data_rest_global_account");
}

Feed MarketDataClient::QueryTakerBuySellRatio(
    const std::string& mea_str,
    const std::string& norm,
    TradingDataInterval interval,
    int64_t start,
    int64_t end) const {
  return QueryTradingData(mea_str, norm, interval, start, end, "trading_data_rest_taker");
}

Feed MarketDataClient::QueryOpenInterestHist(
    const std::string& mea_str,
    const std::string& norm,
    TradingDataInterval interval,
    int64_t start,
    int64_t end) const {
  return QueryTradingData(mea_str, norm, interval, start, end, "trading_data_rest_open_interest");
}

Feed MarketDataClient::QueryTradingData(
    const std::string& mea_str,
    const std::string& norm,
    TradingDataInterval interval,
    int64_t start,
    int64_t end,
    const std::string& channel) const {
  start = start / 1'000'000'000 * 1'000'000'000;
  end = end / 1'000'000'000 * 1'000'000'000;
  CHECK_GT(end, start) << "Invalid time duration " << start << "-" << end;
  TradingDataFeedReader trading_data_reader(start, end);
  auto mea = MarketExchangeApi::FromString(mea_str);
  std::string interval_channel;
  CHECK(TradingDataIntervalChannelConverter(&interval, &interval_channel));
  interval_channel = channel + "_" + interval_channel;
  TradingDataCheckInterval(ci_, mea, interval_channel);
  std::string start_str = Iso8601FromTimestamp(start);
  std::string end_str = Iso8601FromTimestamp(end);
  auto config = BuildConfig(mea_str, norm, interval_channel, start_str, end_str, ci_, nullptr, "trading_data");
  auto prod = BuildProduct(mea_str, config, start);
  CHECK(prod) << "Invliad symbol " << norm;
  auto subscriber = BuildSubscriber(mea, ci_, config);
  auto processor = BuildProcessor(
      ci_, config.feed().exchanges().begin()->first, mea, config);
  RegisterProduct(mea, prod, subscriber, processor, &trading_data_reader);
  auto connection_configs = subscriber->GenerateConnectionConfigs();
  SendRequest(mea, subscriber, config.feed().exchanges().begin()->second.websocket(),
      connection_configs, processor);

  return trading_data_reader.GetFeed();
}

}  // coin2::exchange::base::feed::rest
